package app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import utils.MyKafkaUtil;

import java.time.Duration;
//数据流：web/app ->Mysql -> Maxwell -> Kafka(ODS) -> FlinkApp -> Kafka(DWD)
//程 序: Mock -> Mysql -> Maxwell -> Kafka(ZK) -> Dwd03_trade_order_detail -> Kafka(ZK)

public class Dwd03_trade_order_detail {
    public static void main(String[] args) {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
        //  状态后端设置
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                10, Time.of(1L, TimeUnit.DAYS), Time.of(3L, TimeUnit.MINUTES)
//        ));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
//        System.setProperty("HADOOP_USER_NAME", "atguigu");
        //2.使用flinksql的方式kafka ODS层读取topic_db的数据
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("trade_cart_add"));
        //3.筛选出订单明细表order_detail的数据
        Table orderDetailTable = tableEnv.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['course_id'] course_id,\n" +
                "`data`['course_name'] course_name,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['origin_amount'] origin_amount,\n" +
                "`data`['coupon_reduce'] coupon_reduce,\n" +
                "`data`['final_amount'] final_amount,\n" +
                "`data`['session_id'] session_id\n" +
                "from topic_db\n" +
                "where `database`='edu'\n" +
                "and `table`='order_detail'\n" +
                "and `type`='insert'" );
        tableEnv.createTemporaryView("order_detail",orderDetailTable);
        //4.筛选出订单表order_info的数据
        Table orderInfoTable = tableEnv.sqlQuery("select \n" +
                "`data`['id'] id,\n" +
                "`data`['origin_amount'] origin_amount,\n" +
                "`data`['coupon_reduce'] coupon_reduce,\n" +
                "`data`['final_amount'] final_amount,\n" +
                "`data`['order_status'] order_status,\n" +
                "`data`['out_trade_no'] out_trade_no,\n" +
                "`data`['trade_body'] trade_body,\n" +
                "`data`['session_id'] session_id,\n" +
                "`data`['province_id'] province_id,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['expire_time'] expire_time,\n" +
                "`data`['update_time'] update_time\n" +
                "from topic_db\n" +
                "where `database`='edu'\n" +
                "and `table`='order_info'\n" +
                "and `type`='insert'" );
        tableEnv.createTemporaryView("order_info",orderInfoTable);
        //5.关联两张表
        Table orderDetailResult = tableEnv.sqlQuery("select \n" +
                "od.id,\n" +
                "od.order_id,\n" +
                "od.user_id,\n" +
                "od.course_id,\n" +
                "od.course_name,\n" +
                "od.create_time,\n" +
                "od.origin_amount,\n" +
                "od.coupon_reduce,\n" +
                "od.final_amount,\n" +
                "od.session_id,\n" +
                "oi.order_status,\n" +
                "oi.out_trade_no,\n" +
                "oi.trade_body,\n" +
                "oi.province_id\n" +
                "from order_detail od\n" +
                "join order_info oi\n" +
                "on od.order_id=oi.id" );
        tableEnv.createTemporaryView("trade_order_detail",orderDetailResult);
        //6.通过普通kafka创建sink表
        tableEnv.executeSql("create table dwd_trade_order_detail(\n" +
                "id string,\n" +
                "order_id string,\n" +
                "user_id string,\n" +
                "course_id string,\n" +
                "course_name string,\n" +
                "create_time string,\n" +
                "origin_amount string,\n" +
                "coupon_reduce string,\n" +
                "final_amount string,\n" +
                "session_id string,\n" +
                "order_status string,\n" +
                "out_trade_no string,\n" +
                "trade_body string,\n" +
                "province_id string\n" +
                ")"+MyKafkaUtil.getKafkaSinkConnOption("dwd_trade_order_detail"));
        //7.写入数据
        tableEnv.executeSql("insert into dwd_trade_order_detail select * from trade_order_detail");
    }
}
